/*
  +----------------------------------------------------------------------+
  | PHP Version 5                                                        |
  +----------------------------------------------------------------------+
  | Copyright (c) 1997-2006 George Schlossnagle, The PHP Group           |
  +----------------------------------------------------------------------+
  | This source file is subject to version 3.01 of the PHP license,      |
  | that is bundled with this package in the file LICENSE, and is        |
  | available through the world-wide-web at the following url:           |
  | http://www.php.net/license/3_01.txt.                                 |
  | If you did not receive a copy of the PHP license and are unable to   |
  | obtain it through the world-wide-web, please send a note to          |
  | license@php.net so we can mail you a copy immediately.               |
  +----------------------------------------------------------------------+
  | Author: George Schlossnagle <george@omniti.com>                      |
  |         Pierre-A. Joye <pierre@php.net>                              |
  |         Rob Richards <rrichards@php.net>                             |
  +----------------------------------------------------------------------+
*/
/* $Id: php_spread.c,v 1.54 2008/10/22 16:23:29 rrichards Exp $ */

//#11111

#include "config.h"

#include "php_spread.h"
#include "php_globals.h"
#include "php.h"
#include "ext/standard/php_string.h"
#include "ext/standard/info.h"

#include <sp.h>
#include <sp_func.h>
#ifdef HAVE_SYS_TIME_H
#include <sys/time.h>
#endif

#include <errno.h>

#define ACK fprintf(stderr, "%s:%d\n", __FILE__, __LINE__)

/* for function declarations */

/* This comes from php install tree */
#include "ext/standard/info.h"

static int le_conn;

/* {{{ arginfo */
static
ZEND_BEGIN_ARG_INFO_EX(arginfo_spread_connect, 0, 0, 1)
	ZEND_ARG_INFO(0, spread_daemon)
	ZEND_ARG_INFO(0, private_name)
	ZEND_ARG_INFO(0, receive_joins)
ZEND_END_ARG_INFO()

ZEND_BEGIN_ARG_INFO_EX(arginfo_spread_multicast, 0, 0, 3)
	ZEND_ARG_INFO(0, id)
	ZEND_ARG_INFO(0, group)
	ZEND_ARG_INFO(0, message)
	ZEND_ARG_INFO(0, service_type)
	ZEND_ARG_INFO(0, message_type)
ZEND_END_ARG_INFO()

ZEND_BEGIN_ARG_INFO_EX(arginfo_spread_disconnect, 0, 0, 1)
	ZEND_ARG_INFO(0, id)
ZEND_END_ARG_INFO()

ZEND_BEGIN_ARG_INFO_EX(arginfo_spread_join, 0, 0, 2)
	ZEND_ARG_INFO(0, id)
	ZEND_ARG_INFO(0, group)
ZEND_END_ARG_INFO()

ZEND_BEGIN_ARG_INFO_EX(arginfo_spread_receive, 0, 0, 1)
	ZEND_ARG_INFO(0, id)
	ZEND_ARG_INFO(0, timeout)
ZEND_END_ARG_INFO()

ZEND_BEGIN_ARG_INFO_EX(arginfo_spread___construct, 0, 0, 0)
ZEND_END_ARG_INFO()

ZEND_BEGIN_ARG_INFO_EX(arginfo_spread_oop_connect, 0, 0, 1)
	ZEND_ARG_INFO(0, spread_daemon)
	ZEND_ARG_INFO(0, private_name)
	ZEND_ARG_INFO(0, receive_joins)
ZEND_END_ARG_INFO()

ZEND_BEGIN_ARG_INFO_EX(arginfo_spread_oop_multicast, 0, 0, 2)
	ZEND_ARG_INFO(0, group)
	ZEND_ARG_INFO(0, message)
	ZEND_ARG_INFO(0, service_type)
	ZEND_ARG_INFO(0, message_type)
ZEND_END_ARG_INFO()

ZEND_BEGIN_ARG_INFO_EX(arginfo_spread_oop_disconnect, 0, 0, 0)
ZEND_END_ARG_INFO()

ZEND_BEGIN_ARG_INFO_EX(arginfo_spread_oop_join, 0, 0, 1)
	ZEND_ARG_INFO(0, group)
ZEND_END_ARG_INFO()

ZEND_BEGIN_ARG_INFO_EX(arginfo_spread_oop_receive, 0, 0, 0)
	ZEND_ARG_INFO(0, timeout)
ZEND_END_ARG_INFO()
/* }}} */


/* {{{ spread_functions[] */

function_entry spread_functions[] = {
	PHP_FE(spread_connect, arginfo_spread_connect)
	PHP_FE(spread_multicast, arginfo_spread_multicast)
	PHP_FE(spread_disconnect, arginfo_spread_disconnect)
	PHP_FE(spread_join, arginfo_spread_join)
	PHP_FE(spread_leave, arginfo_spread_join)
	PHP_FE(spread_receive, arginfo_spread_receive)
	{NULL, NULL, NULL}
};
/* }}} */

#ifdef ZEND_ENGINE_2
zend_class_entry *spread_class_entry;
static zend_object_handlers spread_object_handlers;

/* {{{ spread_fetch_object()
 */
static inline ze_spread_object *spread_fetch_object(zval *object TSRMLS_DC)
{
	return (ze_spread_object *) zend_object_store_get_object(object TSRMLS_CC);
}
/* }}} */

/* {{{ spread_object_free_storage */
static void spread_object_free_storage(void *object TSRMLS_DC)
{
	ze_spread_object * intern = (ze_spread_object *) object;

	int i;

	if (!intern) {
		return;
	}

	if (intern->mbox) {
		SP_disconnect(*intern->mbox);
		efree(intern->mbox);
		intern->mbox = NULL;
	}

	if (intern->zo.properties) {
		zend_hash_destroy(intern->zo.properties);
		FREE_HASHTABLE(intern->zo.properties);
	}

	efree(intern);
}
/* }}} */

/* {{{ spread_object_new */
PHP_SPREAD_API  zend_object_value spread_object_new(zend_class_entry *class_type TSRMLS_DC)
{
	ze_spread_object *intern;
	zval *tmp;
	zend_object_value retval;

	intern = emalloc(sizeof(ze_spread_object));
	memset(&intern->zo, 0, sizeof(zend_object));

	intern->zo.ce = class_type;

	intern->mbox = 0;

	ALLOC_HASHTABLE(intern->zo.properties);
	zend_hash_init(intern->zo.properties, 0, NULL, ZVAL_PTR_DTOR, 0);
	zend_hash_copy(intern->zo.properties, &class_type->default_properties, (copy_ctor_func_t) zval_add_ref, (void *) &tmp, sizeof(zval *));

	retval.handle = zend_objects_store_put(intern, NULL, (zend_objects_free_object_storage_t) spread_object_free_storage, NULL TSRMLS_CC);

	retval.handlers = (zend_object_handlers *) & spread_object_handlers;

	return retval;
}
/* }}} */

/* {{{ spread_class_functions */
static zend_function_entry spread_class_functions[] = {
	ZEND_MALIAS(Spread, __construct, spread_constructor,	arginfo_spread___construct,  ZEND_ACC_PUBLIC)
	PHP_ME_MAPPING(connect,         spread_connect,         arginfo_spread_oop_connect,  0)
	PHP_ME_MAPPING(disconnect,      spread_disconnect,      arginfo_spread_oop_disconnect,  0)
	PHP_ME_MAPPING(join,            spread_join,            arginfo_spread_oop_join,  0)
	PHP_ME_MAPPING(leave,           spread_leave,           arginfo_spread_oop_join,  0)
	PHP_ME_MAPPING(receive,         spread_receive,         arginfo_spread_oop_receive,  0)
	PHP_ME_MAPPING(multicast,       spread_multicast,       arginfo_spread_oop_multicast,  0)
	{NULL, NULL, NULL, 0}
};
/* }}} */
#endif

#ifdef ZEND_ENGINE_2
/* {{{ REGISTER_ZIP_CLASS_CONST_LONG */
#define REGISTER_SPREAD_CLASS_CONST_LONG(const_name, value) \
	zend_declare_class_constant_long(spread_class_entry, const_name, sizeof(const_name)-1, (long)value TSRMLS_CC);
/* }}} */
#endif

/* list of exported functions */
/* {{{ module entry */
zend_module_entry spread_module_entry = {
	STANDARD_MODULE_HEADER,
	"spread",
	spread_functions,
	PHP_MINIT(spread),
	PHP_MSHUTDOWN(spread),
	NULL,
	NULL,
	PHP_MINFO(spread),
	PHP_SPREAD_VERSION,
	STANDARD_MODULE_PROPERTIES
};

/* }}} */

#ifdef COMPILE_DL_SPREAD
ZEND_GET_MODULE(spread)
#endif

/* {{{ UTILITIES */

static void php_spread_rsr_dtor(zend_rsrc_list_entry *rsrc TSRMLS_DC) /* {{{ */
{
	mailbox *mbox = (mailbox *)rsrc->ptr;

	if (mbox) {
		SP_disconnect(*mbox);
		efree(mbox);
		rsrc->ptr = NULL;
	}
}

static void php_spread_zend_hash_del_value(HashTable *ht, char *value, size_t value_len TSRMLS_DC)
{
	HashPosition pos;
	zval **tmp;

	zend_hash_internal_pointer_reset_ex(ht, &pos);

	while (zend_hash_get_current_data_ex(ht, (void**) &tmp, &pos) == SUCCESS) {
		if (Z_STRLEN_PP(tmp) == value_len && 0 == strncmp(Z_STRVAL_PP(tmp), value, value_len)) {
			ulong idx;

			if (HASH_KEY_IS_LONG == zend_hash_get_current_key_ex(ht, NULL, NULL, &idx, 0, &pos)) {
				zend_hash_index_del(ht, idx);
				zend_hash_internal_pointer_reset_ex(ht, &pos);
				continue;
			}
		}

		zend_hash_move_forward_ex(ht, &pos);
	}
}

/* }}} */

/* initialization file support */

/* {{{ MINIT */
PHP_MINIT_FUNCTION(spread)
{
#if ZEND_EXTENSION_API_NO >= 220051025 
	int i;

	zend_class_entry spread_ce;

	memcpy(&spread_object_handlers, zend_get_std_object_handlers(), sizeof(zend_object_handlers));

	spread_object_handlers.clone_obj      = NULL;

	INIT_CLASS_ENTRY(spread_ce, "Spread", spread_class_functions);
	spread_ce.create_object = spread_object_new;

	spread_class_entry = zend_register_internal_class(&spread_ce TSRMLS_CC);

	REGISTER_SPREAD_CLASS_CONST_LONG("ILLEGAL_SPREAD",		ILLEGAL_SPREAD);
	REGISTER_SPREAD_CLASS_CONST_LONG("COULD_NOT_CONNECT",	COULD_NOT_CONNECT);
	REGISTER_SPREAD_CLASS_CONST_LONG("CONNECTION_CLOSED",	CONNECTION_CLOSED);
	REGISTER_SPREAD_CLASS_CONST_LONG("REJECT_VERSION",		REJECT_VERSION);
	REGISTER_SPREAD_CLASS_CONST_LONG("REJECT_NO_NAME",		REJECT_NO_NAME);
	REGISTER_SPREAD_CLASS_CONST_LONG("REJECT_ILLEGAL_NAME", REJECT_ILLEGAL_NAME);
	REGISTER_SPREAD_CLASS_CONST_LONG("REJECT_NOT_UNIQUE",	REJECT_NOT_UNIQUE);

	REGISTER_SPREAD_CLASS_CONST_LONG("LOW_PRIORITY",		LOW_PRIORITY);
	REGISTER_SPREAD_CLASS_CONST_LONG("MEDIUM_PRIORITY", 	MEDIUM_PRIORITY);
	REGISTER_SPREAD_CLASS_CONST_LONG("HIGH_PRIORITY",		HIGH_PRIORITY);

	REGISTER_SPREAD_CLASS_CONST_LONG("UNRELIABLE_MESS",		UNRELIABLE_MESS);
	REGISTER_SPREAD_CLASS_CONST_LONG("RELIABLE_MESS",		RELIABLE_MESS);
	REGISTER_SPREAD_CLASS_CONST_LONG("FIFO_MESS",			FIFO_MESS);
	REGISTER_SPREAD_CLASS_CONST_LONG("CAUSAL_MESS",			CAUSAL_MESS);
	REGISTER_SPREAD_CLASS_CONST_LONG("AGREED_MESS",			AGREED_MESS);
	REGISTER_SPREAD_CLASS_CONST_LONG("SAFE_MESS",			SAFE_MESS);
	REGISTER_SPREAD_CLASS_CONST_LONG("REGULAR_MESS",		REGULAR_MESS);

	REGISTER_SPREAD_CLASS_CONST_LONG("REG_MEMB_MESS",			REG_MEMB_MESS);
	REGISTER_SPREAD_CLASS_CONST_LONG("REJECT_MESS",				REJECT_MESS);
	REGISTER_SPREAD_CLASS_CONST_LONG("TRANSITION_MESS",			TRANSITION_MESS);
	REGISTER_SPREAD_CLASS_CONST_LONG("CAUSED_BY_JOIN",			CAUSED_BY_JOIN);
	REGISTER_SPREAD_CLASS_CONST_LONG("CAUSED_BY_LEAVE",			CAUSED_BY_LEAVE);
	REGISTER_SPREAD_CLASS_CONST_LONG("CAUSED_BY_DISCONNECT",	CAUSED_BY_DISCONNECT);
	REGISTER_SPREAD_CLASS_CONST_LONG("CAUSED_BY_NETWORK",		CAUSED_BY_NETWORK);
	REGISTER_SPREAD_CLASS_CONST_LONG("MEMBERSHIP_MESS",			MEMBERSHIP_MESS);
#else
	REGISTER_LONG_CONSTANT("ILLEGAL_SPREAD",		ILLEGAL_SPREAD,			CONST_CS | CONST_PERSISTENT);
	REGISTER_LONG_CONSTANT("COULD_NOT_CONNECT",		COULD_NOT_CONNECT,		CONST_CS | CONST_PERSISTENT);
	REGISTER_LONG_CONSTANT("CONNECTION_CLOSED",		CONNECTION_CLOSED,		CONST_CS | CONST_PERSISTENT);
	REGISTER_LONG_CONSTANT("REJECT_VERSION",		REJECT_VERSION,			CONST_CS | CONST_PERSISTENT);
	REGISTER_LONG_CONSTANT("REJECT_NO_NAME",		REJECT_NO_NAME,			CONST_CS | CONST_PERSISTENT);
	REGISTER_LONG_CONSTANT("REJECT_ILLEGAL_NAME",	REJECT_ILLEGAL_NAME, 	CONST_CS | CONST_PERSISTENT);
	REGISTER_LONG_CONSTANT("REJECT_NOT_UNIQUE",		REJECT_NOT_UNIQUE,		CONST_CS | CONST_PERSISTENT);

	REGISTER_LONG_CONSTANT("SP_LOW_PRIORITY",		LOW_PRIORITY,			CONST_CS | CONST_PERSISTENT);
	REGISTER_LONG_CONSTANT("SP_MEDIUM_PRIORITY",	MEDIUM_PRIORITY,		CONST_CS | CONST_PERSISTENT);
	REGISTER_LONG_CONSTANT("SP_HIGH_PRIORITY",		HIGH_PRIORITY,			CONST_CS | CONST_PERSISTENT);

	REGISTER_LONG_CONSTANT("SP_UNRELIABLE_MESS",	UNRELIABLE_MESS,		CONST_CS | CONST_PERSISTENT);
	REGISTER_LONG_CONSTANT("SP_RELIABLE_MESS",		RELIABLE_MESS,			CONST_CS | CONST_PERSISTENT);
	REGISTER_LONG_CONSTANT("SP_FIFO_MESS",			FIFO_MESS,				CONST_CS | CONST_PERSISTENT);
	REGISTER_LONG_CONSTANT("SP_CAUSAL_MESS",		CAUSAL_MESS,			CONST_CS | CONST_PERSISTENT);
	REGISTER_LONG_CONSTANT("SP_AGREED_MESS",		AGREED_MESS,			CONST_CS | CONST_PERSISTENT);
	REGISTER_LONG_CONSTANT("SP_SAFE_MESS",			SAFE_MESS,				CONST_CS | CONST_PERSISTENT);
	REGISTER_LONG_CONSTANT("SP_REGULAR_MESS",		REGULAR_MESS,			CONST_CS | CONST_PERSISTENT);

	REGISTER_LONG_CONSTANT("SP_REG_MEMB_MESS",		REG_MEMB_MESS,			CONST_CS | CONST_PERSISTENT);
	REGISTER_LONG_CONSTANT("SP_REJECT_MESS",		REJECT_MESS,			CONST_CS | CONST_PERSISTENT);
	REGISTER_LONG_CONSTANT("SP_TRANSITION_MESS",	TRANSITION_MESS,		CONST_CS | CONST_PERSISTENT);
	REGISTER_LONG_CONSTANT("SP_CAUSED_BY_JOIN",		CAUSED_BY_JOIN,			CONST_CS | CONST_PERSISTENT);
	REGISTER_LONG_CONSTANT("SP_CAUSED_BY_LEAVE",	CAUSED_BY_LEAVE,		CONST_CS | CONST_PERSISTENT);
	REGISTER_LONG_CONSTANT("SP_CAUSED_BY_DISCONNECT", CAUSED_BY_DISCONNECT, CONST_CS | CONST_PERSISTENT);
	REGISTER_LONG_CONSTANT("SP_CAUSED_BY_NETWORK",	CAUSED_BY_NETWORK,		CONST_CS | CONST_PERSISTENT);
	REGISTER_LONG_CONSTANT("SP_MEMBERSHIP_MESS",	MEMBERSHIP_MESS,		CONST_CS | CONST_PERSISTENT);
#endif
	le_conn = zend_register_list_destructors_ex(php_spread_rsr_dtor, NULL, "spread", module_number);
	return SUCCESS;
}
/* }}} */

/* {{{ MSHUTDOWN */
PHP_MSHUTDOWN_FUNCTION(spread)
{
	return SUCCESS;
}
/* }}} */

/* {{{ MINFO */
PHP_MINFO_FUNCTION(spread)
{
	char sp_version[128];
	int     mver, miver, pver;

	php_info_print_table_start();
	{
		php_info_print_table_row(2, "Spread", "enabled");
		php_info_print_table_row(2, "Version", PHP_SPREAD_VERSION);
		php_info_print_table_row(2, "CVS Id", "$Id: php_spread.c,v 1.54 2008/10/22 16:23:29 rrichards Exp $");
		if (SP_version( &mver, &miver, &pver)) {
			snprintf(sp_version, 128, "%d.%d.%d", mver, miver, pver);
			php_info_print_table_row(2, "Spread version", sp_version);
		}
	}
	php_info_print_table_end();
}
/* }}} */


#ifdef ZEND_ENGINE_2
/* {{{ proto void Spread::__construct()
   */
PHP_METHOD(Spread, spread_constructor)
{
	zval *object = getThis();
	zval *groups_array;

	add_property_bool(object, "connected", 0);
	add_property_null(object, "daemon");
	add_property_null(object, "private_name");
	add_property_null(object, "private_group");

	MAKE_STD_ZVAL(groups_array);
	array_init(groups_array);
	add_property_zval(object, "joined_groups", groups_array);

	/* add_prop_zval increments the refcount, useless here */
	zval_ptr_dtor(&groups_array);
}
#endif


/* {{{  int _connect(char *spread_name, char *private_name TSRMLS_DC)
 * Open a persistent connection to a spread daemon.  Returns resource id# */
static mailbox *_phpspread_connect(char *spread_name, char *private_name, zend_bool receive_joins, char *private_group TSRMLS_DC)
{
	mailbox *mbox;
	int retval;

	mbox = emalloc(sizeof(mailbox));

	retval = SP_connect(spread_name, private_name, 0, receive_joins, mbox, private_group);

	if (retval != ACCEPT_SESSION) {
		php_error_docref(NULL TSRMLS_CC, E_WARNING, "Failed to connect to spread daemon (%s) using private_name (%s), error returned was: %d", spread_name, private_name, retval);
		efree(mbox);
		return 0;
	}

	return mbox;
}
/* }}} */

/* {{{ proto int spread_connect(string spread_daemon [, string private_name, [bool receive_joins]])
  Open a persistent connection to a spread daemon.  Returns resource id# */
PHP_FUNCTION(spread_connect)
{
	char *spread_name = NULL, *private_name = NULL;
	int spread_name_len, private_name_len;
	zend_bool receive_joins = 0;
	mailbox *mbox = NULL;
	char private_group[MAX_GROUP_NAME];
	int rsrc_id;
	list_entry *le;

#ifdef ZEND_ENGINE_2
	zval *this = getThis();
	ze_spread_object *obj;
	if (this) {
		obj = spread_fetch_object(this TSRMLS_CC);
	}
#endif

	if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "s|sb", &spread_name, &spread_name_len, &private_name, &private_name_len, &receive_joins) == FAILURE) {
		return;
	}

	if (!private_name) {
		char default_private_name[MAX_PRIVATE_NAME];

		snprintf(default_private_name, MAX_PRIVATE_NAME, "php-%05d", getpid());
		private_name = default_private_name;
	} else {
		if (private_name_len > MAX_PRIVATE_NAME) {
			RETURN_LONG(REJECT_ILLEGAL_NAME);
		}
	}

	mbox = _phpspread_connect(spread_name, private_name, receive_joins, private_group TSRMLS_CC);
	if (!mbox) {
		RETURN_FALSE;
	}

#ifdef ZEND_ENGINE_2
	if (this) {
		obj->mbox = mbox;

		zend_update_property_stringl(spread_class_entry, this, "daemon", sizeof("daemon") - 1, spread_name, spread_name_len TSRMLS_CC);
		zend_update_property_stringl(spread_class_entry, this, "private_name", sizeof("private_name") - 1, private_name, strlen(private_name) TSRMLS_CC);
		zend_update_property_stringl(spread_class_entry, this, "private_group", sizeof("private_group") - 1, private_group, strlen(private_group) TSRMLS_CC);
		zend_update_property_bool(spread_class_entry, this, "connected", sizeof("connected") - 1, 1 TSRMLS_CC);

		RETURN_TRUE;
	} else
#endif
	{
		/* hint: next lines sets return value */
		ZEND_REGISTER_RESOURCE(return_value, mbox, le_conn);
	}
}
/* }}} */

/* {{{ proto int spread_multicast(resource id, zval group, string message [, int service_type [, int message_type]])
  Send message to one or more groups */
PHP_FUNCTION(spread_multicast) {
	zval *group;
	zval *optional_mbox;
	int *mbox;
	char *message;
	int sperrno, message_len;
	long service_type = RELIABLE_MESS;
	long mess_type = 1;

#ifdef ZEND_ENGINE_2
	zval *this = getThis();

	if (this) {
		ze_spread_object *obj;

		if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "zs|ll",  &group, &message, &message_len, &service_type, &mess_type) == FAILURE) {
			return;
		}

		obj = spread_fetch_object(this TSRMLS_CC);
		mbox = obj->mbox;
	} else
#endif
	{
		if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "rzs|ll",  &optional_mbox, &group, &message, &message_len, &service_type, &mess_type) == FAILURE) {
			return;
		}

		ZEND_FETCH_RESOURCE(mbox, mailbox *, &optional_mbox, -1, "Spread-FD", le_conn);
	}

	if (! mbox) {
		php_error_docref(NULL TSRMLS_CC, E_WARNING, "Invalid Spread Connection.");
		RETURN_FALSE;
	}
	
	if (mess_type < SHRT_MIN || mess_type > SHRT_MAX) {
		php_error_docref(NULL TSRMLS_CC, E_WARNING, "Message type value out of range.");
		RETURN_FALSE;
	}

	if (Z_TYPE_P(group) == IS_STRING) {
		sperrno = SP_multicast(*mbox, (int)service_type, Z_STRVAL_P(group), (int16) mess_type, message_len, message);
		if (sperrno < 0) {
			php_error_docref(NULL TSRMLS_CC, E_WARNING, "SP_mulicast error(%d)", sperrno);
			RETURN_FALSE;
		}
		RETURN_LONG(sperrno);
	} else if (Z_TYPE_P(group) == IS_ARRAY) {
		char groupnames[100][MAX_GROUP_NAME];
		HashPosition pos;
		zval *tmparr, **tmp;
		int n = 0;

		zend_hash_internal_pointer_reset_ex(group->value.ht, &pos);
		while(zend_hash_get_current_data_ex(group->value.ht, (void **) &tmp, &pos) == SUCCESS && n < 100) {
			convert_to_string_ex(tmp);
			strncpy(groupnames[n], Z_STRVAL_PP(tmp), MAX_GROUP_NAME);
			n++;
			zend_hash_move_forward_ex(group->value.ht, &pos);
		}
		sperrno = SP_multigroup_multicast(*mbox, (int)service_type, n, (const char (*)[MAX_GROUP_NAME]) groupnames, (int16) mess_type, message_len, message);
		if (sperrno < 0) {
			php_error_docref(NULL TSRMLS_CC, E_WARNING, "SP_mulicast error(%d)", sperrno);
			RETURN_FALSE;
		}

		RETURN_LONG(sperrno);
	} else {
		php_error_docref(NULL TSRMLS_CC, E_WARNING, "SP_Join: expect groups to an array of strings or a string (got: %d)", Z_TYPE_P(group));
		RETURN_FALSE;
	}
}
/* }}} */

/* {{{ proto int spread_join(resource id, zval group)
  Join one or more groups */
PHP_FUNCTION(spread_join) {
	zval *group;
	zval **joined_groups = NULL;
	mailbox *mbox;
	int sperrno;

#ifdef ZEND_ENGINE_2
	zval *this = getThis();
	
	if (this) {
		ze_spread_object *obj;

		if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "z", &group) == FAILURE) {
			return;
		}

		obj = spread_fetch_object(this TSRMLS_CC);
		if (zend_hash_find(obj->zo.properties, "joined_groups", sizeof("joined_groups"), (void **)&joined_groups) == FAILURE) {
			RETURN_FALSE;
		}
		mbox = obj->mbox;
	} else
#endif
	{
		zval *mbox_zval;
		if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "rz", &mbox_zval, &group) == FAILURE) {
			return;
		}
		ZEND_FETCH_RESOURCE(mbox, mailbox *, &mbox_zval, -1, "spread", le_conn);
	}

	if (!mbox) {
		php_error_docref(NULL TSRMLS_CC, E_WARNING, "Resource or mailbox connection invalid.");
		RETURN_FALSE;
	}

	if (Z_TYPE_P(group) == IS_STRING) {
		if ( (sperrno = SP_join(*mbox, Z_STRVAL_P(group))) < 0) {
			RETURN_LONG(sperrno);
		}

#ifdef ZEND_ENGINE_2
		if (joined_groups != NULL) {
			add_next_index_stringl(*joined_groups, Z_STRVAL_P(group), Z_STRLEN_P(group), 1);
		}
#endif
		RETURN_TRUE;
	} else if (Z_TYPE_P(group) == IS_ARRAY) {
		char groupnames[100][MAX_GROUP_NAME];
		HashPosition pos;
		HashTable *tmparr_hash;
		zval **tmp;
		int error = 0, n = 0;

		tmparr_hash = Z_ARRVAL_P(group);

		if (zend_hash_num_elements(tmparr_hash) < 1) {
			RETURN_TRUE;
		}

		for (zend_hash_internal_pointer_reset_ex(tmparr_hash, &pos);
				zend_hash_get_current_data_ex(tmparr_hash,(void**) &tmp, &pos) == SUCCESS && (n++ <= 100);
				zend_hash_move_forward_ex(tmparr_hash, &pos)) {

			convert_to_string_ex(tmp);
			if (Z_STRLEN_PP(tmp) < 1 || Z_STRLEN_PP(tmp) > MAX_GROUP_NAME) {
				RETURN_LONG(ILLEGAL_GROUP);
			}

			if ((sperrno = SP_join(*mbox,  Z_STRVAL_PP(tmp))) < 0) {
				error = sperrno;
				break;
			}

#ifdef ZEND_ENGINE_2
			if (joined_groups != NULL) {
				add_next_index_stringl(*joined_groups, Z_STRVAL_PP(tmp), Z_STRLEN_PP(tmp), 1);
			}
#endif
		}

		if (error) {
			RETURN_LONG(sperrno);
		}

		RETURN_TRUE;
	} else {
		php_error_docref(NULL TSRMLS_CC, E_WARNING, "expect groups to be an array of strings or a string (got: %d)", Z_TYPE_P(group));
		RETURN_LONG(ILLEGAL_GROUP);
	}
}
/* }}} */

/* {{{ proto int spread_leave(resource id, zval group)
  Leave one or more groups */
PHP_FUNCTION(spread_leave) {
	zval *group;
	zval **joined_groups = NULL;
	mailbox *mbox;
	int sperrno;

#ifdef ZEND_ENGINE_2
	zval *this = getThis();
	
	if (this) {
		ze_spread_object *obj;

		if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "z", &group) == FAILURE) {
			return;
		}

		obj = spread_fetch_object(this TSRMLS_CC);
		if (zend_hash_find(obj->zo.properties, "joined_groups", sizeof("joined_groups"), (void **)&joined_groups) == FAILURE) {
			RETURN_FALSE;
		}
		mbox = obj->mbox;
	} else
#endif
	{
		zval *mbox_zval;
		if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "rz", &mbox_zval, &group) == FAILURE) {
			return;
		}
		ZEND_FETCH_RESOURCE(mbox, mailbox *, &mbox_zval, -1, "spread", le_conn);
	}

	if (!mbox) {
		php_error_docref(NULL TSRMLS_CC, E_WARNING, "Resource or mailbox connection invalid.");
		RETURN_FALSE;
	}

	if (Z_TYPE_P(group) == IS_STRING) {
		if ( (sperrno = SP_leave(*mbox, Z_STRVAL_P(group))) < 0) {
			RETURN_LONG(sperrno);
		}

#ifdef ZEND_ENGINE_2
		if (joined_groups != NULL) {
			php_spread_zend_hash_del_value(Z_ARRVAL_PP(joined_groups), Z_STRVAL_P(group), Z_STRLEN_P(group) TSRMLS_CC);
		}
#endif
		RETURN_TRUE;
	} else if (Z_TYPE_P(group) == IS_ARRAY) {
		char groupnames[100][MAX_GROUP_NAME];
		HashPosition pos;
		HashTable *tmparr_hash;
		zval **tmp;
		int error = 0, n = 0;

		tmparr_hash = Z_ARRVAL_P(group);

		if (zend_hash_num_elements(tmparr_hash) < 1) {
			RETURN_TRUE;
		}

		for (zend_hash_internal_pointer_reset_ex(tmparr_hash, &pos);
				zend_hash_get_current_data_ex(tmparr_hash,(void**) &tmp, &pos) == SUCCESS && (n++ <= 100);
				zend_hash_move_forward_ex(tmparr_hash, &pos)) {

			convert_to_string_ex(tmp);
			if (Z_STRLEN_PP(tmp) < 1 || Z_STRLEN_PP(tmp) > MAX_GROUP_NAME) {
				RETURN_LONG(ILLEGAL_GROUP);
			}

			if ((sperrno = SP_leave(*mbox,  Z_STRVAL_PP(tmp))) < 0) {
				error = sperrno;
				break;
			}

#ifdef ZEND_ENGINE_2
			if (joined_groups != NULL) {
				php_spread_zend_hash_del_value(Z_ARRVAL_PP(joined_groups), Z_STRVAL_PP(tmp), Z_STRLEN_PP(tmp) TSRMLS_CC);
			}
#endif
		}

		if (error) {
			RETURN_LONG(sperrno);
		}

		RETURN_TRUE;
	} else {
		php_error_docref(NULL TSRMLS_CC, E_WARNING, "expect groups to be an array of strings or a string (got: %d)", Z_TYPE_P(group));
		RETURN_LONG(ILLEGAL_GROUP);
	}
}
/* }}} */

/* {{{ proto int spread_receive(resource id [,int timeout])
  Receive a message */
PHP_FUNCTION(spread_receive) {
	zval *groups_zval;
	int id = -1;
	int *mbox;
	double timeout = 0;
	struct timeval towait;
	int sperrno;
	int i, endmis, ret, ngrps, msize=0;
	int16 mtype;
	service stype = 0;
	int oldmsize = 0;
	int oldgsize = 0;
	int newmsize = (1<<15);
	int newgsize = (1<<6);
	char* groups=NULL;
	char* mess=NULL;
	char sender[MAX_GROUP_NAME];
	fd_set readfs;

#ifdef ZEND_ENGINE_2
	zval *this = getThis();

	if (this) {
		ze_spread_object *obj;

		if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "|d", &timeout) == FAILURE) {
			return;
		}

		obj = spread_fetch_object(this TSRMLS_CC);
		mbox = obj->mbox;
	} else
#endif
	{ 
		zval *mbox_zval;
		if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "r|d", &mbox_zval, &timeout) == FAILURE) {
			return;
		}
		ZEND_FETCH_RESOURCE(mbox, mailbox *, &mbox_zval, id, "Spread-FD", le_conn);
	}

	if (!mbox) {
		php_error_docref(NULL TSRMLS_CC, E_WARNING, "Resource or mailbox connection invalid.");
		RETURN_FALSE;
	}

	FD_ZERO(&readfs);
	FD_SET(*mbox, &readfs);

	if (-1 == timeout) {
		ret = select(*mbox+1, &readfs, NULL, &readfs, NULL);
	} else {
		towait.tv_sec = (unsigned long)timeout;
		towait.tv_usec = (unsigned long)(1000000.0 * (timeout - (double)towait.tv_sec));
		ret = select(*mbox+1, &readfs, NULL, &readfs, &towait);
	}

	if (ret < 1) {
		if (-1 == ret) {
			php_error_docref(NULL TSRMLS_CC, E_WARNING, "errno = %d", errno);
		} else if (ret == 0) {
			RETURN_NULL();
		}

		RETURN_FALSE;
	}

try_again:
	{
		if (oldgsize != newgsize) {
			if (groups) {
				groups = (char*) erealloc(groups, newgsize * MAX_GROUP_NAME);
			} else {
				groups = (char*) emalloc(newgsize * MAX_GROUP_NAME);
			}
			oldgsize=newgsize;
		}
		if (oldmsize != newmsize) {
			if (mess) {
				mess = (char *) erealloc(mess, newmsize);
			} else {
				mess = (char *) emalloc(newmsize);
			}
			oldmsize = newmsize;
		}

		ret = SP_receive(*mbox, &stype, sender, newgsize, &ngrps, 
			(char (*)[MAX_GROUP_NAME])groups, &mtype, &endmis, newmsize, mess);

		if (ret < 0) {
			/* some error */
			if (ret == BUFFER_TOO_SHORT) {
				newmsize = -endmis;
				newmsize++;
				goto try_again;
			}
		} else {
			if (newmsize != (ret + 1)) {
				mess = (char *) erealloc(mess, ret + 1);
			}
		}
		msize = ret;
	}

	if (ret < 0 || array_init(return_value) == FAILURE) {
		if (groups) {
			efree(groups);
		}
		if (mess) {
			efree(mess);
		}

		if (ret < 0) {
			php_error_docref(NULL TSRMLS_CC, E_WARNING, "spread returned %d", ret);
		}

		RETURN_FALSE;
	}

	add_assoc_long(return_value, "service_type", stype);
	if (Is_membership_mess(stype)) {
		/* For membership-messages, "sender" field contains group, in which event occured */
		add_assoc_string(return_value, "group", sender, 1);

		if (Is_reg_memb_mess(stype)) {
			/* join/leave/disconnect/network */
			membership_info memb_info;
			char *changed_member;

			add_assoc_long(return_value, "group_size", ngrps);

			MAKE_STD_ZVAL(groups_zval);
			array_init(groups_zval);
			if (groups) {
				for(i = 0; i < ngrps; i++) {
					add_index_stringl(groups_zval, i, &groups[i*MAX_GROUP_NAME], strlen(&groups[i*MAX_GROUP_NAME]), 1);
				}

				efree(groups);
			}
			add_assoc_zval(return_value, "group_members", groups_zval);
			add_assoc_long(return_value, "current_index", mtype);

			/* getting additional info */
#if (SPREAD_VERSION >> 24) >= 4
			SP_get_memb_info(mess, stype, &memb_info);
			changed_member = memb_info.changed_member;
#else
			changed_member = &mess[SP_get_vs_set_offset_memb_mess()];
#endif

			if (Is_caused_join_mess(stype)) {
				add_assoc_string(return_value, "changed_member", changed_member, 1);
			} else if (Is_caused_leave_mess(stype)) {
				add_assoc_string(return_value, "changed_member", changed_member, 1);
			} else if (Is_caused_disconnect_mess(stype)) {
				add_assoc_string(return_value, "changed_member", changed_member, 1);
			} else if (Is_caused_network_mess(stype)) {
				/* TODO */
			}

			efree(mess);
		} else if (Is_transition_mess(stype)) {
			/* transition */
			/* There is no additional info, for this kind of messages */
			if (groups) {
				efree(groups);
			}
			if (mess) {
				efree(mess);
			}
		} else if (Is_self_leave(stype)) {
			/* self leave */
			
		} else {
			php_error_docref(NULL TSRMLS_CC, E_WARNING, "Service-type %d is unhandled. File bug-report please", stype);
		}
	} else {
		/* regular message */
		add_assoc_stringl(return_value, "message", mess, msize, 0);

		MAKE_STD_ZVAL(groups_zval);
		array_init(groups_zval);
		if (groups) {
			for(i = 0; i < ngrps; i++) {
				add_index_stringl(groups_zval, i, &groups[i*MAX_GROUP_NAME], strlen(&groups[i*MAX_GROUP_NAME]), 1);
			}

			efree(groups);
		}
		add_assoc_zval(return_value, "groups", groups_zval);
		add_assoc_long(return_value, "message_type", mtype);
		add_assoc_string(return_value, "sender", sender, 1);
	}
}
/* }}} */

/* {{{ proto int spread_close(resource id)
   Close a spread connection */
PHP_FUNCTION(spread_disconnect) {
	zval *spread_conn;
	mailbox *mbox;
	int retval;

#ifdef ZEND_ENGINE_2
	zval **joined_groups = NULL;
	zval *this = getThis();
	if (this) {
		ze_spread_object *obj = spread_fetch_object(this TSRMLS_CC);

		zend_hash_find(obj->zo.properties, "joined_groups", sizeof("joined_groups"), (void **)&joined_groups);

		mbox = obj->mbox;
		if (mbox) {
			int retval;

			retval = SP_disconnect(*mbox);
			efree(mbox);
			mbox = NULL;
			obj->mbox = NULL;

			zend_update_property_bool(spread_class_entry, this, "connected", sizeof("connected") - 1, 0 TSRMLS_CC);
			zend_update_property_null(spread_class_entry, this, "daemon", sizeof("daemon") - 1 TSRMLS_CC);
			zend_update_property_null(spread_class_entry, this, "private_name", sizeof("private_name") - 1 TSRMLS_CC);
			zend_update_property_null(spread_class_entry, this, "private_group", sizeof("private_group") - 1 TSRMLS_CC);
			zend_hash_clean(Z_ARRVAL_PP(joined_groups));

			if (retval == 0) {
				RETURN_TRUE;
			} else {
				RETURN_LONG(retval);
			}
		} else {
			php_error_docref(NULL TSRMLS_CC, E_WARNING, "Invalid connection.");
			RETURN_FALSE;
		}
	} else 
#endif
	{
		if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "r", &spread_conn) == FAILURE) {
			return;
		}

		/* Automatically returns FALSE, in case of error */
		ZEND_FETCH_RESOURCE(mbox, mailbox *, &spread_conn, -1, "Spread-FD", le_conn);

		retval = SP_disconnect(*mbox);

		zend_list_delete(Z_RESVAL_P(spread_conn));
		if (retval == 0) {
			RETURN_TRUE;
		} else {
			RETURN_LONG(retval);
		}
	}
}
/* }}} */
